/*
 * Bulletin board server by Pedro Tabacof - ID 2101353
 * CSC414 - Computer Networks - Prof. Stefan Bruda
 * Bishop's University
 *
 * This is a simple bulletin board multi-threaded server.
 *
 * More information can be found in the README file or in the comments below.
 */

#include "tcp-utils.h"
#include "thread-utils.h"

#define T_MAX 10 //Maximum number of active threads.
#define T_INCR 2 //Number of threads that are (de)allocated together.
#define MAX_CHAR 512 //Maximum size of a board entry.
#define TIMEOUT 50 //Value in milliseconds of the timeout of the polling function in each slave thread.

FILE *fp; //Master pointer to the file.
//Mutexes are used to protect some variables containing file access and thread information.
pthread_mutex_t file_count_mutex, thread_count_mutex;
pthread_cond_t file_count_threshold_cv;
//Various counters that are protected by the mutexes above.
int file_rw_count = 0, active_thread_count = 0, allocated_thread_count = 0, message_count = 0;
int n; //This variable will usually contain allocated_thread_count - active_thread_count.
pthread_t thread_pool[T_MAX]; //Thread pool of inactive threads.
char file_name[50]; //Bulletin board file name.


/* Retrieves a message of identifier 'number' from the bulletin board. */
void read_msg (int sd, int number)
{
    char read_msg[MAX_CHAR+1]; //String containing message that will be sent to the client.
    char line[MAX_CHAR+1], msg[MAX_CHAR+1], *toks; //Auxiliary strings.
    int rd_line, fp_read, flag = 0, error = 0;

    //Increases the number of threads reading from the file.
    pthread_mutex_lock(&file_count_mutex);
    file_rw_count++;
    pthread_mutex_unlock(&file_count_mutex);

    fp_read = open(file_name, O_RDONLY); //New file pointer so it can read without intereference.
    if (readline(fp_read, line, MAX_CHAR) < 0)   //Jump over message count
    {
        error++; //Error reading line makes the function stop
        sprintf(read_msg, "ERROR could not read line from file.\n");
    }

    while (error == 0) //Reads until the end of the file.
    {
        rd_line = readline(fp_read, line, MAX_CHAR); //Reads a line
        if (rd_line == recv_nodata)
            break;
        else if (rd_line < 0) //Error reading line makes the function stop
        {
            error++;
            sprintf(read_msg, "ERROR could not read line from file.\n");
            break;
        }
        toks = strtok(line, "/"); //Gets message number
        if (number == atoi(toks)) //If found a message matching the number, reads it into msg,
        {
            flag++;
            toks = strtok(NULL, "\n");
            strncpy(msg, toks, MAX_CHAR);
        }
    }
    if (error > 0) //In case there was an error reading a line, send proper information.
    {
        send(sd, read_msg, strlen(read_msg), 0);
        file_rw_count--;
        if (file_rw_count == 0)
            pthread_cond_signal(&file_count_threshold_cv);
        pthread_mutex_unlock(&file_count_mutex);
        return;
    }
    if (flag == 0) //If no matching number was found...
        sprintf(read_msg, "UNKNOWN message #%d could not find the specified message.\n", number);
    else
        sprintf(read_msg, "MESSAGE %d %s\n", number, msg);

    send(sd, read_msg, strlen(read_msg), 0);

    close(fp_read);

    //Decreases reading count and sends a condition variable signal in case file is available for writing operations.
    pthread_mutex_lock(&file_count_mutex);
    file_rw_count--;
    if (file_rw_count == 0) //Sends signal to threads that may want to write on file.
        pthread_cond_signal(&file_count_threshold_cv);
    pthread_mutex_unlock(&file_count_mutex);
}

/* Writes a message from the client in the bulletin board and returns its identifier. */
void write_msg (int sd, char *text, char *user_name)
{
    char write_msg[50]; //Message that will be sent to the client.

    //Can only access file for writing if noone is reading or writing on it.
    //If blocked, it will wait until a conditional variable signal is sent
    //from the last thread to read or write in the file.
    pthread_mutex_lock(&file_count_mutex);
    if (file_rw_count != 0)
        pthread_cond_wait(&file_count_threshold_cv, &file_count_mutex);
    message_count++; //Increases message count
    fseek(fp, 0, SEEK_SET);
    if (fprintf(fp, "%10d\n", message_count) < 0) //If cant update message count, the function fails.
    {
        sprintf(write_msg, "ERROR WRITE couldnt update message count on file.\n");
        message_count--;
        send(sd, write_msg, strlen(write_msg), 0);
        pthread_mutex_unlock(&file_count_mutex);
        return;
    }
    fseek(fp, 0, SEEK_END);
    if (fprintf(fp, "%d/%s/%s\n", message_count, user_name, text) < 0) //If cant write message, the function fails.
    {
        sprintf(write_msg, "ERROR WRITE couldnt write message on the end of the file.\n");
        send(sd, write_msg, strlen(write_msg), 0);
        pthread_mutex_unlock(&file_count_mutex);
        return;
    }
    fflush(fp); //Flushes content to file.
    //Releases mutex lock and also sends signal informing its free to write on now.
    pthread_cond_signal(&file_count_threshold_cv);
    pthread_mutex_unlock(&file_count_mutex);
    //Send information to the client.
    sprintf(write_msg, "WROTE message #%d\n", message_count);
    send(sd, write_msg, strlen(write_msg), 0);
}

/* This function replaces a bulletin board entry by another.
 * Its design is a mix of two functions, read_msg and write_msg,
 * so it is advised to look at them first.
 * OBS: The message is not actually replaced, but a new copy of it
 * will appear at the end of the file. Since read_msg always gets
 * the last message posted, it acts as if the message has been replaced
 * and a rollback function becomes easier to be implemented.
 */
void replace_msg (int sd, char *text, char *user_name, int number)
{
    char replace_msg[MAX_CHAR+1]; //Message that will replace the old one.
    char line[MAX_CHAR+1], *toks; //Auxiliary strings
    int rd_line, fp_read, error = 0, flag = 0;

    //Can only access file for writing if noone is reading or writing on it.
    //If blocked, it will wait until a conditional variable signal is sent
    //from the last thread to read or write in the file.
    pthread_mutex_lock(&file_count_mutex);
    if (file_rw_count != 0)
        pthread_cond_wait(&file_count_threshold_cv, &file_count_mutex);

    fp_read = open(file_name, O_RDONLY); //File pointer used to find message to be replaced.
    if (readline(fp_read, line, MAX_CHAR) < 0)   //Jump over message count
    {
        error++;
        sprintf(replace_msg, "ERROR could not read line from file.\n");
    }
    while (error == 0) //Tries to find entry in the board with the same message number.
    {
        rd_line = readline(fp_read, line, MAX_CHAR);
        if (rd_line == recv_nodata) //End of file
            break;
        else if (rd_line < 0) //Error
        {
            error++;
            sprintf(replace_msg, "ERROR could not read line from file.\n");
            break;
        }
        toks = strtok(line, "/");
        if (number == atoi(toks)) //Found an entry with the specified number
        {
            flag++;
        }
    }
    close(fp_read);
    if (flag == 0 && error == 0) //If couldnt find the message...
    {
        sprintf(replace_msg, "UNKNOWN message #%d could not find the specified message.\n", number);
    }
    if (flag == 0 || error > 0) //In case of error or inexistent message, the function fails.
    {
        send(sd, replace_msg, strlen(replace_msg), 0);
        pthread_cond_signal(&file_count_threshold_cv);
        pthread_mutex_unlock(&file_count_mutex);
        return;
    }

    //Now it is time to write new message at the end of the file.
    fseek(fp, 0, SEEK_END);
    if (fprintf(fp, "%d/%s/%s\n", number, user_name, text) < 0)
    {
        sprintf(replace_msg, "ERROR WRITE couldnt write message on the end of the file.\n");
        send(sd, replace_msg, strlen(replace_msg), 0);
        pthread_mutex_unlock(&file_count_mutex);
        return;
    }
    fflush(fp);
    //Releases mutex lock and also sends signal informing its free to write on now.
    pthread_cond_signal(&file_count_threshold_cv);
    pthread_mutex_unlock(&file_count_mutex);
    sprintf(replace_msg, "WROTE message #%d\n", number);
    send(sd, replace_msg, strlen(replace_msg), 0);
}

/* This is the function executed by the threads that will handle connection directly with client
 * and satisfy its requests.
 */
void *slave_thread(void *ms)
{
    int sd; //Slave socket for this thread
    char message[MAX_CHAR+1]; //Message received by the server from the client
    struct sockaddr_in client_addr; //The address of the client
    socklen_t client_addr_len = sizeof(client_addr); //Client's length
    int msock = (int) ms; //Master socket
    char *toks; //Used to tokenize a line.
    char user_name[25] = "nobody"; //User name of the client.
    char bye[80]; //Quiting message.
    char greeting[60]; //Greeting message.
    char text[MAX_CHAR+1];
    struct pollfd fds;
    const char unknown[] = "ERROR unknown command - please refer to README file.\n";
    int poll_result;
    int garbage;

    //This is done in order for the threads to be cancelled while blocked.
    //Otherwise for a not understood reason a deadlock would occur.
    //Since cancellation occurs only when the monitor has the mutex, the mutex is safe.
    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, &garbage);

    while (1) //We now go into an infinite loop in a thread...
    {
        fds.revents = 0;
	fds.events = POLLIN;
	fds.fd = msock;

	/* Locks thread pool so this thread will have exlusivity with any client connecting in the next
         * TIMEOUT milliseconds. Also makes it impossible for this thread to be reaped when a client connects,
         * because the monitor also needs to acquire this lock.
         * The lock is not released until a connection is found or timeout happens because otherwise
         * if the scheduler changes working thread before that happens, this thread would be considered
         * active even if it is not. Thus, after a round of scheduling all threads would appear to be busy
         * by the monitor even if they are not, and the monitor would allocate more threads.
         */
        pthread_mutex_lock(&thread_count_mutex);
        n = allocated_thread_count - active_thread_count;
        active_thread_count++;
        thread_pool_remove(pthread_self(), thread_pool, n);

        if ((poll_result = poll(&fds, (nfds_t) 1, TIMEOUT)) > 0) //Polls for incoming connection for TIMEOUT ms.
            sd = accept(msock, (struct sockaddr*)&client_addr, &client_addr_len); //If there is a pending connection, accepts it.

        //If no connection was found or if there was an error looking for or accepting them,
        //this thread is put in the inactive thread pool and the lock is released, so now
        //another thread may look for a client or the monitor may do its job.
        if (poll_result <= 0 || sd < 0) {
	    n = allocated_thread_count - active_thread_count;
            active_thread_count--;
            thread_pool_insert(pthread_self(), thread_pool, n);
            pthread_mutex_unlock(&thread_count_mutex);
            sched_yield(); //Thread yield
	    continue;
        }

        pthread_mutex_unlock(&thread_count_mutex); //Now that the thread is not in the inactive thread pool

        // Loop while the client has something to say...
        while ((n = readline(sd, message, MAX_CHAR-1)) != recv_nodata)
        {
            toks = strtok(message, " \n\r");
            if (toks == NULL) //No information is received.
                continue;
            else if (!strncmp(toks, "quit", MAX_CHAR) || !strncmp(toks, "QUIT", MAX_CHAR)) //Client wants to quit.
            {
                sprintf(bye, "BYE %s thank you for your visit!\n", user_name);
                send(sd, bye, strlen(bye), 0);
                break;
            }
            else if (!strncmp(toks, "user", MAX_CHAR) || !strncmp(toks, "USER", MAX_CHAR)) //Client changes user name.
            {
                toks = strtok(NULL, "/\n\r"); //Gets user name parsed for '/' and newlines.
                strncpy(user_name, toks, 25); //No buffer overflow possible.
                user_name[24] = '\0'; //Ensure it is a valid string.
                sprintf(greeting, "HELLO %s welcome to my server!\n", user_name);
                send(sd, greeting, strlen(greeting), 0);
                continue;
            }
            else if (!strncmp(toks, "read", MAX_CHAR) || !strncmp(toks, "READ", MAX_CHAR)) //Client wants to read a message.
            {
                toks = strtok(NULL, " \n\r"); //Gets message number.
                if (toks == NULL) {
			send(sd, "ERROR no message number.\n", 25, 0);
			continue;
		}

		int number = atoi(toks);
                read_msg(sd, number);
            }
            else if (!strncmp(toks, "write", MAX_CHAR) || !strncmp(toks, "WRITE", MAX_CHAR)) //Client wants to write a message.
            {
                int message_len_counter = 0;
                toks = strtok(NULL, " \n\r");
                text[0] = '\0';
                while (toks != NULL) //Gets message parsed for blanks and newlines
                {
                    message_len_counter += strlen(toks); //This is made to avoid possibility of buffer overflow.
                    if (message_len_counter >= MAX_CHAR)
                        break;
                    strncat(text, toks, MAX_CHAR-1);
                    strcat(text, " ");
                    toks = strtok(NULL, " \n\r");
                }
                text[strlen(text)-1] = '\0'; //Ensures message is a valid string.
                text[MAX_CHAR-1] = '\0';

                write_msg(sd, text, user_name);
            }

            else if (!strncmp(toks, "replace", MAX_CHAR) || !strncmp(toks, "REPLACE", MAX_CHAR)) //Client wants to replace a message.
            {
                toks = strtok(NULL, " \n\r"); //Gets message number
                int number = atoi(toks);

                toks = strtok(NULL, " \n\r"); //Gets client message
                text[0] = '\0';
                while (toks != NULL)
                {
                    strncat(text, toks, MAX_CHAR-1);
                    strcat(text, " ");
                    toks = strtok(NULL, " \n\r");
                }
                text[strlen(text)-1] = '\0'; //Ensures the message is a valid string.
                text[MAX_CHAR-1] = '\0';

                replace_msg(sd, text, user_name, number);
            }
            else //Unknown command.
            {
                send(sd, unknown, strlen(unknown), 0);
                continue;
            }
        }
        //If client quits or disconnects
        strcpy(user_name, "nobody");
        shutdown(sd,1);
        close(sd);

        //Updates the number of active threads and inserts this thread into the inactive thread pool.
        pthread_mutex_lock(&thread_count_mutex);
        n = allocated_thread_count - active_thread_count;
        active_thread_count--;
        thread_pool_insert(pthread_self(), thread_pool, n);
        pthread_mutex_unlock(&thread_count_mutex);
        sched_yield(); //Thread yield

    }
    return NULL;
}

/* This function monitor all slave threads. It it is needed, it allocates or dealloactes
 * a batch of T_INCR threads.
 * There will always be at least T_INCR threads allocated and never more than T_INCR + T_MAX.
 * If the number of active threads is T_INCR + 1 less than the number of allocated threads,
 * this thread will reap T_INCR threads only if all conditions above are satisfied.
 */
void *thread_monitor(void *ms)
{
    pthread_t threads; //Thread number
    pthread_attr_t attr; //Thread attribute
    int i; //Loop variable
    int msock = (int) ms; //Master socket

    /* Initialize attributes so that the threads will be dettached. */
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);

    allocated_thread_count = T_INCR;

    //Allocates the initial batch of T_INCR threads.
    for (i = 0; i < T_INCR; i++)
    {
        if ( pthread_create(&threads, &attr, slave_thread, (void*)msock) != 0 )
        {
            printf("Monitor: error creating a thread.\n");
            allocated_thread_count--;
        }
        thread_pool_insert(threads, thread_pool, i+1);
    }

    while (1) //Loops forever watching the thread count every 60 seconds.
    {
        pthread_mutex_lock(&thread_count_mutex); //Locks thread count related variables.
        printf("Monitor: %d allocated threads, %d active threads.\n", allocated_thread_count, active_thread_count);
        if (active_thread_count == allocated_thread_count) //If more threads need to be allocated.
        {
            if (allocated_thread_count <= T_MAX)
            {
                allocated_thread_count += T_INCR;
                //Allocates T_INCR more threads.
                printf("Monitor: allocating %d threads.\n", T_INCR);
                for (i = 0; i < T_INCR; i++)
                {
                    if ( pthread_create(&threads, &attr, slave_thread, (void*)msock) != 0 )
                    {
                        printf("Monitor: error creating a thread.\n");
                        allocated_thread_count--;
                    }
                    thread_pool_insert(threads, thread_pool, i+1); //All threads are initially inactive.
                }
            }
        }
        //If threads need to be dealloacted.
        else if ((allocated_thread_count > T_INCR) && (allocated_thread_count - active_thread_count > T_INCR))
        {
            n = allocated_thread_count - active_thread_count;
            //Deallocates T_INCR threads.
            printf("Monitor: deallocating %d threads.\n", T_INCR);
            for (i = 0; i < T_INCR; i++)
            {
                if (pthread_cancel(thread_pool[i]))
                {
                    printf("Monitor: error deallocating thread #%d - server will have to shut down.\n", (int)thread_pool[i]);
                    exit(-1);
                }
                thread_pool_remove(thread_pool[i], thread_pool, n-i); //Remove them from thread_pool.
            }
            allocated_thread_count = allocated_thread_count - T_INCR;
        }

        pthread_mutex_unlock(&thread_count_mutex); //Unlocks mutex - other threads now may become active or inactive.
        sleep(5); //Sleeps for one minute.
    }
    pthread_exit(NULL);
    return NULL;
}

int main (int argc, char *argv[])
{
    int master_socket;
    pthread_t monitor; //Monitor thread that will spawn or kill slave threads
    pthread_attr_t attr; //Thread attribute
    int errn, port;

    if (argc < 2)   //There must be at least one parameter, that is, the file name of the bulletin board.
    {
        printf("Insufficient parameters - please refer to README file.\n");
        exit(-1);
    }

    if (argc >= 3) //If the user provided a port as well, uses it instead of the default (8000)
        port = atoi(argv[2]);
    else
        port = 8000;

    strncpy(file_name, argv[1], 50);
    file_name[49] = '\0';

    //Opens the file that contains all bulletin board entries - creates it if it does not exist.
    fp = fopen(file_name, "r+");
    if (fp == NULL)
    {
        fp = fopen(file_name, "w+");
        if (fp == NULL)
        {
            printf("Unable to open file %s.\n", argv[1]);
            exit(-1);
        }
        fprintf(fp, "%10d\n", message_count); //Message count is initialized as zero.
    }
    else
    {
        fseek(fp, 0, SEEK_SET);
        fscanf(fp, "%d", &message_count);
    }

    /* Initialize mutex and condition variable objects */
    pthread_mutex_init(&file_count_mutex, NULL);
    pthread_mutex_init(&thread_count_mutex, NULL);
    pthread_cond_init (&file_count_threshold_cv, NULL);

    /* Initialize attributes so that the threads will be dettached. */
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);

    master_socket = passivesocket(port, 10); //Creates a master socket for incoming connections.
    if (master_socket < 0)
    {
        printf("Error #%d creating master socket. \n", master_socket);
        exit(-1);
    }
    
    /* Master socket is set as non-blockable so no deadlock regarding the mutex can occur. */
    int flags;
    flags = fcntl(master_socket, F_GETFL, 0);
    if (flags == -1 || (fcntl(master_socket, F_SETFL, flags | O_NONBLOCK) == -1)) {
	printf("Error configuring the socket.\n");
	exit(-1);
    } 

    errn = pthread_create(&monitor, &attr, thread_monitor, (void *)master_socket); //Creates monitor thread - it spawns and kills slave threads if necessary.
    if (errn)
    {
        printf("Error #%d creating monitor thread - program will exit.\n", errn);
        exit(-1);
    }

   //Server is now functioning - no need for main thread anymore. 
    while(1)
        sleep(1000);

    return -1; //Should never get here.
}
